package com.runa.monitor.platinterface.netservice.connect;

import com.runa.monitor.platinterface.base.BaseContainer;
import com.runa.monitor.platinterface.netservice.connect.commstatus.CommStatus;
import com.runa.monitor.platinterface.netservice.equipment.PlcService;
import com.runa.monitor.platinterface.netservice.server.ListenerServer;
import com.runa.protocol.dto.IInterData;
import com.runa.protocol.dto.IResponseData;
import io.netty.channel.Channel;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collection;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 旧实现，已废弃，只供参考
 * @author zhuj
 */
@Deprecated
public class ListenerSession implements Session {

    private static final Logger logger = LoggerFactory.getLogger(ListenerServer.class);

    private Channel channel;

    private boolean online;

    private Queue<IRequest> requestQueue = new ConcurrentLinkedQueue<>();

    private IRequest currRequest;

    private PlcService service;

    private volatile boolean key = true;

    private int timeOut;

    private long prevTime;

    private AtomicInteger flag = new AtomicInteger(0);

    public ListenerSession(PlcService service, int timeout) {
        this.service = service;
        this.timeOut = timeout == 0 ? 1000 :timeout;
    }

    @Override
    public void linked(Channel channel) {
        if (channel != null && channel.isActive()) {
            this.channel = channel;
            BaseContainer.putSession(channel.id().asLongText(), this);
            setOnlineStatus(true);
        }
    }

    public synchronized void pushRequest(Collection<IRequest> requests) {
        int f = flag.addAndGet(1);
        for (IRequest request : requests) {
            request.flag(f);
            requestQueue.offer(request);
        }
    }

    @Override
    public void setOnlineStatus(boolean status) {
        this.online = status;
        this.service.updateCommStatus(status);
    }

    @Override
    public boolean isOnline() {
        return online;
    }

    @Override
    public void requesting() {
        if (!online) {
            currRequest = requestQueue.poll();
            if (currRequest != null) {
                reset(CommStatus.OffLine);
            }
        } else {
            if (key) {
                currRequest = requestQueue.poll();
                if (currRequest != null) {
                    channel.writeAndFlush(currRequest.getRequestData());
                    prevTime = System.currentTimeMillis();
                    key = false;
                }
            } else if (System.currentTimeMillis() - prevTime > timeOut) { //超时机制
                reset(CommStatus.TimeOut);
            }
        }
    }

    private void reset(CommStatus status) {
        logger.info("Session Reset PLC:{} --- {}", service.getId(), status);
        key = true;
        while (requestQueue.peek() != null &&
                requestQueue.peek().flag(1) == currRequest.flag(1)) {
            requestQueue.poll();
        }
        currRequest.cancel(status);
    }

    @Override
    public void receiving(IResponseData data) {
        if (!key) {
            IResponse<IInterData> response = currRequest.getResponse();
            response.setResponseData(data);
            if (currRequest.check()){
                key = true;
                response.doReceive();
            }
        }
    }

    @Override
    public boolean close() {
        try {
            if (isOnline() && channel != null) {
                AttributeKey key = AttributeKey.valueOf("isActiveClose");
                channel.attr(key);
                channel.close();
            }
        } catch (Exception e) {
            return false;
        } finally {
            setOnlineStatus(false);
        }
        return true;
    }

}
